package com.atguigu.tingshu.album.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.album.service.AlbumStatService;
import com.atguigu.tingshu.common.anno.GuiguCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.AuthContextHolder;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	AlbumAttributeValueMapper albumAttributeValueMapper;

	@Autowired
	AlbumStatMapper albumStatMapper;

	@Autowired
	AlbumStatService albumStatService;

	@Autowired
	AlbumAttributeValueService albumAttributeValueService;

	@Autowired
	KafkaService kafkaService;

	@Autowired
	RedissonClient redissonClient;

	@Autowired
	StringRedisTemplate stringRedisTemplate;

	@Override
	public IPage<AlbumListVo> findUserAlbumPage(Long pageNo, Long pageSize, AlbumInfoQuery albumInfoQuery) {
		//前端可以不用传用户Id
		albumInfoQuery.setUserId(AuthContextHolder.getUserId());
		IPage<AlbumListVo> page = new Page(pageNo, pageSize);
		return albumInfoMapper.selectUserAlbumPage(page,albumInfoQuery);
	}
	@Transactional
	@Override
	public void removeAlbumInfoById(Long id) {
		//删除专辑表的数据 album_info
		this.removeById(id);
		//删除专辑属性信息
		albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,id));
		//删除专辑对应的统计数据
		albumStatMapper.delete(new LambdaQueryWrapper<AlbumStat>().eq(AlbumStat::getAlbumId, id));
		//查询数据库专辑基本数据，如果专辑的状态为公开，则删除ES中的数据，即下架
		if("1".equals(getAlbumInfoById(id).getIsOpen())){
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(getAlbumInfoById(id).getId()));
		}
	}

	@GuiguCache(cacheKey = RedisConstant.ALBUM_INFO_PREFIX,enableLock = true,lockName = RedisConstant.ALBUM_LOCK_SUFFIX)
	@Override
	public AlbumInfo getAlbumInfoById(Long id) {
				//从数据库中查询
		return getAlbumInfoFromDB(id);
	}

	//封装从数据库查询数据的方法
	public AlbumInfo getAlbumInfoFromDB(Long id){
		AlbumInfo albumInfo = albumInfoMapper.selectById(id);
		//查询专辑对应的属性值列表，最终返回带有属性值列表的albumInfo
		if(albumInfo !=null){
			List<AlbumAttributeValue> attributeValueList = albumAttributeValueService.list(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, id));
			albumInfo.setAlbumAttributeValueVoList(attributeValueList);
		}
		return albumInfo;
	}

	public AlbumInfo getAlbumInfoByIdRedission(Long id) {
		String redisKey = RedisConstant.ALBUM_INFO_PREFIX + id;
		String albumInfotoJson = stringRedisTemplate.opsForValue().get(redisKey);
		if(!StringUtils.isEmpty(albumInfotoJson)){
			log.info("从redis中查询到了AlbumInfo数据");
			return JSON.parseObject(albumInfotoJson, AlbumInfo.class);
		}
		//获取分布锁
		RLock rLock = redissonClient.getLock(RedisConstant.ALBUM_LOCK_SUFFIX + id);
		if(rLock.tryLock()){ //体现的是限流思想，降低数据库的访问压力
			try {
				log.info(Thread.currentThread().getName()+"获得了锁对象，并且从数据库中开始查询数据。。。" +id);
				//从数据库中查询
				AlbumInfo albumInfoFromDB = getAlbumInfoFromDB(id);
				stringRedisTemplate.opsForValue().set(redisKey, JSON.toJSONString(albumInfoFromDB),7, TimeUnit.DAYS);
				log.info(Thread.currentThread().getName()+"从数据库中查询到数据并将数据放在redis中。。。" +id);
				return albumInfoFromDB;
			}catch (Exception e){
				throw new RuntimeException(e);
			}finally { //为避免业务代码执行过程中程序出错造成本线程持有的所资源无法释放进而产生死锁问题。需将释放所操作放在finally中保证无论代码执行过程中是否出现异常最终都会释放所资源
				log.info(Thread.currentThread().getName()+"业务操作执行完毕并且释放了锁资源。。。" +id);
				rLock.unlock();
			}
		}else {
			log.info(Thread.currentThread().getName()+"未获得锁对象，并自旋继续执行本方法");
			return getAlbumInfoByIdRedission(id);
		}
	}

	@Transactional
	@Override
	public void updateAlbumInfoById(Long id, AlbumInfoVo albumInfoVo) {
		//使用延迟双删保证缓存和数据库的一致性 ->使用更友好的canal技术实现
		//stringRedisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX +id);

		//根据id数据修改info信息
		AlbumInfo albumInfo = this.getById(id);
		String isOpenOld = albumInfo.getIsOpen();
		BeanUtils.copyProperties(albumInfoVo, albumInfo);
		this.updateById(albumInfo);
		// 修改专辑的属性值数据
		// 1、先把之前属性值数据删除掉
		albumAttributeValueService.remove(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, id));

		// 2、新增属性值数据
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
			AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
			albumAttributeValue.setAlbumId(id);
			albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
			albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
			return albumAttributeValue;
		}).collect(Collectors.toList());
		albumAttributeValueService.saveBatch(albumAttributeValueList);
		//根据原专辑的状态和更新后专辑的状态变化，确定上架与下架
		if("1".equals(isOpenOld)){
			//if(!isOpenOld.equals(albumInfoVo.getIsOpen())){
				kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(id));
			//}
		}else {
			//if(!isOpenOld.equals(albumInfoVo.getIsOpen())){
				kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, String.valueOf(id));
			//}
		}
		//执行延迟双删 ->使用更友好的canal技术实现
//		ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(16);
//		scheduledThreadPool.schedule(()->{
//			stringRedisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX +id);
//			log.info("延迟删除任务执行了。。。。");
//		}, 300, TimeUnit.MILLISECONDS);

	}

	@Transactional
	@Override
	public void saveAlbumInfo(AlbumInfoVo albumInfoVo) {
		// 保存专辑的基本数据
		Long userId = AuthContextHolder.getUserId();
		AlbumInfo albumInfo = new AlbumInfo();
		BeanUtils.copyProperties(albumInfoVo, albumInfo);
		albumInfo.setUserId(userId);
		albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
		// 付费专辑需要设置试听的声音的集数
		if(!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfo.getPayType())){
			albumInfo.setTracksForFree(5);
		}
		albumInfoMapper.insert(albumInfo);
		// 保存专辑的属性值数据
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
			AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
			albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
			albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
			albumAttributeValue.setAlbumId(albumInfo.getId());
			return albumAttributeValue;
		}).collect(Collectors.toList());
		albumAttributeValueService.saveBatch(albumAttributeValueList);

		// 初始化专辑统计信息数据
		ArrayList<AlbumStat> albumStatList = new ArrayList<>();
		albumStatList.add(new AlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY, 0));
		albumStatList.add(new AlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE, 0));
		albumStatList.add(new AlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE, 0));
		albumStatList.add(new AlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT, 0));
		albumStatService.saveBatch(albumStatList);

		if("1".equals(albumInfo.getIsOpen())){
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, String.valueOf(albumInfo.getId()));
		}
	}

	@Override
	public List<AlbumInfo> findUserAllAlbumList() {
		Long userId = AuthContextHolder.getUserId();
		LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumInfoLambdaQueryWrapper.eq(AlbumInfo::getUserId, userId );
		albumInfoLambdaQueryWrapper.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle);
		return albumInfoMapper.selectList(albumInfoLambdaQueryWrapper);
	}

	@GuiguCache(cacheKey = RedisConstant.ALBUM_STAT_PREFIX,enableLock = true,lockName = RedisConstant.ALBUM_STAT_LOCK)
	@Override
	public AlbumStatVo findAlbumStatByAlbumId(Long albumId) {
		return albumStatMapper.findAlbumStatByAlbumId(albumId);
	}
}
